/*

 * Licensed to the Apache Software Foundation (ASF) under one

 * or more contributor license agreements.  See the NOTICE file

 * distributed with this work for additional information

 * regarding copyright ownership.  The ASF licenses this file

 * to you under the Apache License, Version 2.0 (the

 * "License"); you may not use this file except in compliance

 * with the License.  You may obtain a copy of the License at

 *

 *     http://www.apache.org/licenses/LICENSE-2.0

 *

 * Unless required by applicable law or agreed to in writing, software

 * distributed under the License is distributed on an "AS IS" BASIS,

 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

 * See the License for the specific language governing permissions and

 * limitations under the License.

 */

package com.bff.gaia.unified.runners.core.construction;



import com.bff.gaia.unified.sdk.coders.Coder;

import com.bff.gaia.unified.sdk.runners.AppliedPTransform;

import com.bff.gaia.unified.sdk.runners.PTransformOverrideFactory;

import com.bff.gaia.unified.sdk.transforms.Create;

import com.bff.gaia.unified.sdk.transforms.Create.Values;

import com.bff.gaia.unified.sdk.transforms.PTransform;

import com.bff.gaia.unified.sdk.values.PBegin;

import com.bff.gaia.unified.sdk.values.PCollection;

import com.bff.gaia.unified.sdk.values.PCollection.IsBounded;

import com.bff.gaia.unified.vendor.guava.com.google.common.collect.Iterables;

import com.bff.gaia.unified.sdk.values.PValue;

import com.bff.gaia.unified.sdk.values.TupleTag;

import com.bff.gaia.unified.sdk.values.WindowingStrategy;



import java.util.Map;



/** An implementation of {@link Create} that returns a primitive {@link PCollection}. */

public class PrimitiveCreate<T> extends PTransform<PBegin, PCollection<T>> {

  private final Create.Values<T> transform;

  private final Coder<T> coder;



  private PrimitiveCreate(Create.Values<T> transform, Coder<T> coder) {

    this.transform = transform;

    this.coder = coder;

  }



  @Override

  public PCollection<T> expand(PBegin input) {

    return PCollection.createPrimitiveOutputInternal(

        input.getPipeline(), WindowingStrategy.globalDefault(), IsBounded.BOUNDED, coder);

  }



  public Iterable<T> getElements() {

    return transform.getElements();

  }



  /** A {@link PTransformOverrideFactory} that creates instances of {@link PrimitiveCreate}. */

  public static class Factory<T>

      implements PTransformOverrideFactory<PBegin, PCollection<T>, Values<T>> {

    @Override

    public PTransformReplacement<PBegin, PCollection<T>> getReplacementTransform(

        AppliedPTransform<PBegin, PCollection<T>, Values<T>> transform) {

      return PTransformReplacement.of(

          transform.getPipeline().begin(),

          new PrimitiveCreate<>(

              transform.getTransform(),

              ((PCollection<T>) Iterables.getOnlyElement(transform.getOutputs().values()))

                  .getCoder()));

    }



    @Override

    public Map<PValue, ReplacementOutput> mapOutputs(

		Map<TupleTag<?>, PValue> outputs, PCollection<T> newOutput) {

      return ReplacementOutputs.singleton(outputs, newOutput);

    }

  }

}